package io.task.cc.message.socket.task;

import io.task.cc.message.socket.*;
import io.task.cc.message.socket.thread.pool.QueueModeSynchronousTask;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.util.Queue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 解码任务对象，一个连接对应一个本对象
 */
@Setter
@Getter
@Slf4j
public class DecodeTask extends QueueModeSynchronousTask<ByteBuffer> {

    private ChannelContext channelContext = null;
    private Config config = null;
    /**
     * 上一次解码剩下的数据
     */
    private ByteBuffer lastByteBuffer = null;
    /**
     * 新收到的数据
     */
    private ByteBuffer newReceivedByteBuffer = null;


    public void handle(Packet packet, int byteCount) {
        /*switch (tioConfig.packetHandlerMode) {
            case SINGLE_THREAD:
                channelContext.handlerRunnable.handler(packet);
                break;
            case QUEUE:
                channelContext.handlerRunnable.addMsg(packet);
                channelContext.handlerRunnable.execute();
                break;
            default:
                channelContext.handlerRunnable.handler(packet);
                break;
        }*/
        channelContext.handlerRunnable.addMsg(packet);
        channelContext.handlerRunnable.execute();
    }


    public DecodeTask(ChannelContext channelContext, Executor executor) {
        super(executor);
        this.channelContext = channelContext;
        this.config = channelContext.config;
        getMsgQueue();
    }

    /**
     * 清空处理的队列消息
     */
    @Override
    public void clearMsgQueue() {
        super.clearMsgQueue();
        lastByteBuffer = null;
        newReceivedByteBuffer = null;
    }

    @Override
    public void runTask() {
        while ((newReceivedByteBuffer = msgQueue.poll()) != null) {
            decode();
        }
    }


    public void decode() {
        ByteBuffer byteBuffer = newReceivedByteBuffer;
        if (lastByteBuffer != null) {
            byteBuffer = ByteBufferUtils.composite(lastByteBuffer, byteBuffer);
            lastByteBuffer = null;
        }

        label_2:
        while (true) {
            try {
                int initPosition = byteBuffer.position();
                int limit = byteBuffer.limit();
                int readableLength = limit - initPosition;
                Packet packet = null;
                if (channelContext.packetNeededLength != null) {
                    if (log.isInfoEnabled()) {
                        log.info("{}, 解码所需长度:{}", channelContext, channelContext.packetNeededLength);
                    }
                    if (readableLength >= channelContext.packetNeededLength) {
                        packet = config.getTioHandler().decode(byteBuffer, limit, initPosition, readableLength, channelContext);
                    }
                } else {
                    try {
                        packet = config.getTioHandler().decode(byteBuffer, limit, initPosition, readableLength, channelContext);
                    } catch (BufferUnderflowException e) {
                        //log.error(e.toString(), e);
                        //数据不够读
                    }
                }

                if (packet == null)// 数据不够，解不了码
                {
                    //					lastByteBuffer = ByteBufferUtils.copy(byteBuffer, initPosition, limit);
                    /*if (tioConfig.useQueueDecode || (byteBuffer != newReceivedByteBuffer)) */
                    {
                        byteBuffer.position(initPosition);
                        byteBuffer.limit(limit);
                        lastByteBuffer = byteBuffer;
                    }/* else {
                        lastByteBuffer = ByteBufferUtils.copy(byteBuffer, initPosition, limit);
                    }*/
                    ChannelState channelState = channelContext.stat;
                    channelState.decodeFailCount++;
                    //					int len = byteBuffer.limit() - initPosition;
                    if (log.isInfoEnabled()) {
                        log.info("{} 本次解码失败, 已经连续{}次解码失败，参与解码的数据长度共{}字节", channelContext, channelState.decodeFailCount, readableLength);
                    }
                    if (channelState.decodeFailCount > 5) {
                        if (channelContext.packetNeededLength == null) {
                            if (log.isInfoEnabled()) {
                                log.info("{} 本次解码失败, 已经连续{}次解码失败，参与解码的数据长度共{}字节", channelContext, channelState.decodeFailCount, readableLength);
                            }
                        }

                        //检查慢包攻击
                        if (channelState.decodeFailCount > 10) {
                            //							int capacity = lastByteBuffer.capacity();
                            int per = readableLength / channelState.decodeFailCount;
                            if (per < Math.min(channelContext.getReadBufferSize() / 2, 256)) {
                                String str = "连续解码" + channelState.decodeFailCount + "次都不成功，并且平均每次接收到的数据为" + per + "字节，有慢攻击的嫌疑";
                                throw new DecodeException(str);
                            }
                        }
                    }
                    return;
                } else //解码成功
                {
                    channelContext.setPacketNeededLength(null);
                    channelContext.stat.latestTimeOfReceivedPacket = System.currentTimeMillis();
                    channelContext.stat.decodeFailCount = 0;

                    int packetSize = byteBuffer.position() - initPosition;
                    packet.setByteCount(packetSize);


                    if (log.isDebugEnabled()) {
                        log.debug("{}, 解包获得一个packet:{}", channelContext);
                    }

                    handle(packet, packetSize);

                    if (byteBuffer.hasRemaining())//组包后，还剩有数据
                    {
                        if (log.isDebugEnabled()) {
                            log.debug("{},组包后，还剩有数据:{}", channelContext, byteBuffer.remaining());
                        }
                        continue label_2;
                    } else//组包后，数据刚好用完
                    {
                        lastByteBuffer = null;
                        if (log.isDebugEnabled()) {
                            log.debug("{},组包后，数据刚好用完", channelContext);
                        }
                        return;
                    }
                }
            } catch (Throwable e) {
                if (channelContext.logWhenDecodeError) {
                    log.error("解码时遇到异常", e);
                }

                channelContext.setPacketNeededLength(null);


                Io.close(channelContext, e, "解码异常:" + e.getMessage(), ChannelContext.CloseCode.DECODE_ERROR);
                return;
            }
        }
    }


    private Queue<ByteBuffer> msgQueue = new LinkedBlockingQueue<>();

    @Override
    public Queue<ByteBuffer> getMsgQueue() {
        return msgQueue;
    }

}
